package message.push.client;

import java.io.IOException;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import message.push.util.Aes;
import message.push.util.MD5;
import message.push.util.TimeProcesser;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBufferInputStream;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.execution.ExecutionHandler;
import org.jboss.netty.handler.execution.OrderedMemoryAwareThreadPoolExecutor;
import org.jboss.netty.util.HashedWheelTimer;
import org.jboss.netty.util.Timer;

import com.cndw.rpg.framework.ResponseMessage;
import com.cndw.rpg.framework.codec.Compress;
import com.cndw.rpg.framework.execution.SimpleThreadFactory;
import com.cndw.rpg.framework.socket.GameServerIdleStateHandler;

import flex.messaging.io.ClassAliasRegistry;
import flex.messaging.io.SerializationContext;
import flex.messaging.io.amf.Amf3Input;

public class SendTest {
	public static SerializationContext context;
	
	private static String loginKey = "tencent-login-8159a693-ebd7-4ec6-8e1d-4384bd6551aa";
	private static String partnerEName = "localhost";
//	private static String partnerEName = "tencent";
	private static String pf = "qzone";
	private static String loginExt = "1,0,openkey,pfkey,0,0,0";
//	private static String loginExt = "1,1,9F0EC056C08DBCF6A7FC1333A20F4B2A,2cbb2847166c567d43a31d956250314b,0,0,0,0";
	
//	private static String openId = "F33E27800D1B02B3C96D1475FF4B5D5A";
	private static String openId = "ch03";
	static {
		context = new SerializationContext();
		context.legacyCollection = true;
		
//		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.rpg.framework.ResponseMessage", "com.cndw.kungfu.socket.SocketResponse");
//		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.rpg.framework.ResponseMessage", "message.push.client.SocketResponse");
//		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.vo.PlayerVO", "message.push.vo.PlayerVO");
//		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.domain.EventNotice", "message.push.vo.EventNotice");
//		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.domain.Player", "message.push.vo.Player");
////		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.vo.UnionVO", "com.cndw.kungfu.socket.vo.Union");
//		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.domain.Notice", "message.push.vo.Notice");
//		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.domain.LogPlayer", "message.push.vo.PlayerLog");
//		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.vo.ModeRecordVO", "message.push.vo.Record");
//		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.vo.PlayerDataVO", "message.push.vo.PlayerData");
//		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.domain.Recharge", "message.push.vo.Recharge");
//		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.domain.Mail", "message.push.vo.Mail");
		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.rpg.framework.ResponseMessage", "com.cndw.kungfu.socket.SocketResponse");
		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.domain.Player", "com.cndw.kungfu.socket.vo.Player");
		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.vo.UnionVO", "com.cndw.kungfu.socket.vo.Union");
		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.domain.Notice", "com.cndw.kungfu.socket.vo.Notice");
		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.domain.LogPlayer", "com.cndw.kungfu.socket.vo.PlayerLog");
		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.vo.ModeRecordVO", "com.cndw.kungfu.socket.vo.Record");
		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.vo.PlayerDataVO", "com.cndw.kungfu.socket.vo.PlayerData");
		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.domain.Recharge", "com.cndw.kungfu.socket.vo.Recharge");
		ClassAliasRegistry.getRegistry().registerAlias("com.cndw.kungfu.domain.Mail", "com.cndw.kungfu.socket.vo.Mail");
	
	}
	
	public static ChannelFuture future ;
	public static ClientBootstrap bootstrap;

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub
//		send();
		sendMsg();
	}

	public static ResponseMessage send() {
		int time = TimeProcesser.getUnixTime();
		Map<String, Object> map = new HashMap<String, Object>();
		map.put("openId", openId);
		map.put("time", time + "");
		map.put("pf", pf);
		String sign = MD5.compile(partnerEName+","+openId+ "," + time +"," + pf +","+ loginExt+"," + loginKey);
		map.put("sign", sign);
		map.put("ext", loginExt);
		ResponseMessage message = null;
		try {
			Socket socket = new Socket();
//			socket.connect(new InetSocketAddress("s10.app100622837.qqopenapp.com", 8001), 2000);
			socket.connect(new InetSocketAddress("localhost", 7782), 2000);
			socket.setSoTimeout(15000);
			socket.setTcpNoDelay(true);
			socket.setKeepAlive(true);
			MySocketRequest loginRequest = new MySocketRequest("LoginController", "auth", map);
			socket.getOutputStream().write(loginRequest.serialize().array());
			socket.getOutputStream().flush();
			Amf3Input input = new Amf3Input(context);
			InputStream stream = null;
			ChannelBuffer buffer = ChannelBuffers.dynamicBuffer();
			int bodySize = 0;
			int compressSize = 0;
			byte bytes[] = new byte[1024];
			ChannelBuffer bodyBuffer = null;
			while (true) {
				bodySize = 0;
				compressSize = 0;
				stream = socket.getInputStream();
				do {
					int length = stream.read(bytes);
					if (length != -1)
						buffer.writeBytes(bytes, 0, length);
					while (bodySize == 0 && buffer.readableBytes() >= 4) {
						bodySize = buffer.readInt();
						if (bodySize > 1024 && compressSize == 0) {
							compressSize = bodySize;
							bodySize = 0;
						}
					}
				} while (buffer.readableBytes() < bodySize);
				
				if (compressSize > 0) {
					byte cpomressBytes[] = Compress.decompressBytes(buffer.readBytes(bodySize).array());
					bodyBuffer = buffer.factory().getBuffer(cpomressBytes.length);
					bodyBuffer.writeBytes(cpomressBytes);
				} else {
					bodyBuffer = buffer.factory().getBuffer(bodySize);
					bodyBuffer.writeBytes(buffer, bodySize);
				}
				input.setInputStream(new ChannelBufferInputStream(bodyBuffer));
				message = (ResponseMessage) input.readObject();
				bodyBuffer.clear();
				System.out.println(message.m);
				input.close();
//				if (socket != null)
//					socket.close();
//				return message;
			}
			
		} catch (IOException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			
		}
		return message;
	}
	
	public static void sendMsg() {
		bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool()));
		bootstrap.setPipelineFactory(new MyChannelPipelineFactory());
		bootstrap.setOption("tcpNoDelay" , true);   
        bootstrap.setOption("keepAlive", true); 
		future = bootstrap.connect(new InetSocketAddress("localhost", 7782));
		Channel channel = future.awaitUninterruptibly().getChannel();
		if (!future.isSuccess()) {
			future.getCause().printStackTrace();
			bootstrap.releaseExternalResources();
			return;
		}
		MySocketRequest loginRequest = new MySocketRequest("LoginController", "auth", getLoginMap());
		future = channel.write(loginRequest.serialize());
		if (future.isSuccess())
//			System.exit(0);
		future.awaitUninterruptibly();
//		bootstrap.releaseExternalResources();
//		KeepAlive.runBoss();
	}
	
	public static void sendIdleMessage () {
		Channel channel = future.awaitUninterruptibly().getChannel();
		if (!future.isSuccess()) {
			future.getCause().printStackTrace();
			bootstrap.releaseExternalResources();
			return;
		}
		MySocketRequest loginRequest = new MySocketRequest("LoginController", "idleMessage");
		future = channel.write(loginRequest.serialize());
	}

	public static String getCipher(String... params) {
		StringBuilder param = new StringBuilder();
		for (String p : params) {
			param.append(p);
			param.append("\r\n");
		}
		param.append(TimeProcesser.getUnixTime());
		return Aes.encrypt(param.toString());
	}
	
	public static Map<String, Object> getLoginMap() {
		int time = TimeProcesser.getUnixTime();
		Map<String, Object> map = new HashMap<String, Object>();
		map.put("openId", openId);
		map.put("time", time + "");
		map.put("pf", pf);
		String sign = MD5.compile(partnerEName+","+openId+ "," + time +"," + pf +","+ loginExt+"," + loginKey);
		map.put("sign", sign);
		map.put("ext", loginExt);
		return map;
	}
	
	// protected static final String openId = "openId";
			// protected static final String time = "time";
			// protected static final String pf = "pf";
			// protected static final String userName = "userName";
			// protected static final String ext = "ext";
			// protected static final String openKey = "openKey";
			// protected static final String pfKey = "pfKey";
			// protected static final String al = "al";
			// protected static final String yYear = "yYear";
			// protected static final String yLevel = "yLevel";
			// protected static final String sign = "sign";
			// protected static final String type = "type";

}

class ClientHandler extends SimpleChannelHandler {

	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
		ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
		ChannelBufferInputStream inputStream = new ChannelBufferInputStream(buffer);
		dealMessage(inputStream);
//		e.getChannel().close();
		
		
//		ChannelBufferInputStream inputStream = new ChannelBufferInputStream((ChannelBuffer) e.getMessage());
//		Amf3Input input = new Amf3Input(SendTest.context);
//		input.setInputStream(inputStream);
//
//		SocketResponse message = (SocketResponse) input.readObject();
//		System.out.println(message.getMessage());
//		MessagePush.pushMessage(message.getMessage());
	}

	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
		System.out.println("Unexpected exception from downstream."
				+ e.getCause());
		e.getChannel().close();
	}
	
	@Override
	public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
		super.channelClosed(ctx, e);
	}

	@Override
	public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
		super.channelDisconnected(ctx, e);
		System.out.println("disconnected...");
//		System.exit(0);
	}

	public ResponseMessage dealMessage(ChannelBufferInputStream inputStream) {
		Amf3Input input = new Amf3Input(SendTest.context);
		InputStream stream = inputStream;
		ChannelBuffer buffer = ChannelBuffers.dynamicBuffer();
		int bodySize = 0;
		int compressSize = 0;
		byte bytes[] = new byte[1024];
		ChannelBuffer bodyBuffer = null;
		ResponseMessage message = null;
		try {
			
			bodySize = 0;
			compressSize = 0;
			do {
				int length = stream.read(bytes);
				if (length != -1)
					buffer.writeBytes(bytes, 0, length);
				while (bodySize == 0 && buffer.readableBytes() >= 4) {
					bodySize = buffer.readInt();
					if (bodySize > 1024 && compressSize == 0) {
						compressSize = bodySize;
						bodySize = 0;
					}
				}
			} while (buffer.readableBytes() < bodySize);
			
			if (compressSize > 0) {
				byte cpomressBytes[] = Compress.decompressBytes(buffer.readBytes(bodySize).array());
				bodyBuffer = buffer.factory().getBuffer(cpomressBytes.length);
				bodyBuffer.writeBytes(cpomressBytes);
			} else {
				bodyBuffer = buffer.factory().getBuffer(bodySize);
				bodyBuffer.writeBytes(buffer, bodySize);
			}
			input.setInputStream(new ChannelBufferInputStream(bodyBuffer));
			message = (ResponseMessage) input.readObject();
			bodyBuffer.clear();
			System.out.println(message.m);
			MessagePush.pushMessage((String)message.m);
			input.close();
//			if (socket != null)
//				socket.close();
//			return message;
			
		} catch (Exception e) {
			e.printStackTrace();
		}
		return message;
	}
}

class KeepAlive implements Runnable {
	
	long checkDelay = 10;
    long keepAliveDelay = 2000;
    
    long lastSendTime;
    public void run() {
        while(true){
            if(System.currentTimeMillis()-lastSendTime > keepAliveDelay){
                try {
                    SendTest.sendIdleMessage();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                lastSendTime = System.currentTimeMillis();
            }else{
                try {
                    Thread.sleep(checkDelay);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
    public static void runBoss() {
    	Thread thread = new Thread(new KeepAlive());
    	thread.start();
    }
    
    

}

class MyChannelPipelineFactory implements ChannelPipelineFactory {
	private static final ExecutionHandler executionHandler;
	private static final Timer timer;

	static {
		java.util.concurrent.ThreadFactory threadFactory = new SimpleThreadFactory("FarmeWorker");
		ThreadPoolExecutor executor = new OrderedMemoryAwareThreadPoolExecutor(500, 0L, 0L, 60L, TimeUnit.SECONDS, threadFactory);
		executor.allowCoreThreadTimeOut(false);
		executionHandler = new ExecutionHandler(executor);
		timer = new HashedWheelTimer(new SimpleThreadFactory("hearbeat"), 1300L, TimeUnit.MILLISECONDS, 512);
	}
	@Override
	public ChannelPipeline getPipeline() throws Exception {
		ChannelPipeline pipeline =  Channels.pipeline();
//		pipeline.addLast("framer", new BaseFrameDecode());
		pipeline.addLast("hearbeat", new GameServerIdleStateHandler(timer, 600, 0, 0));
		pipeline.addLast("handler", new ClientHandler());
		return pipeline;
	}
}
